package vn.com.epi.re.tools;

import java.util.List;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import vn.com.epi.re.ApplicationContext;
import vn.com.epi.re.table.RecommendationTable;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.io.Closeables;

public class CleanRecommendations  extends Configured implements Tool {
    
    public static final int GROUP_SIZE = 1000;

    public static void main(String[] args) throws Exception {
        long startTime = System.currentTimeMillis();
        ApplicationContext c = ApplicationContext.init();
        try {
            ToolRunner.run(new CleanRecommendations(), args);
        } finally {
            Closeables.closeQuietly(c);            
        }        
        long endTime = System.currentTimeMillis();
        System.out.println("total time used: " + (endTime - startTime) + " ms.");
    }

    @Override
    public int run(String[] args) throws Exception {
        HTableInterface table = ApplicationContext.get().getTable(RecommendationTable.NAME);
        try {
            List<Delete> deletes = Lists.newArrayList();
            Scan scan = new Scan();
            scan.addColumn(RecommendationTable.FAM_MAIN, RecommendationTable.COL_RECOMMENDATIONS);        
            ResultScanner rs = table.getScanner(scan);
            int counter = 0;
            for (Result r:rs) {
                String recommendations = Bytes.toString(r.getValue(RecommendationTable.FAM_MAIN, RecommendationTable.COL_RECOMMENDATIONS));
                if (!Strings.isNullOrEmpty(recommendations)) {
                    Delete delete = new Delete(r.getRow());
                    delete.deleteColumn(RecommendationTable.FAM_MAIN, RecommendationTable.COL_RECOMMENDATIONS);
                    deletes.add(delete);
                    counter++;
                }
                if (counter == GROUP_SIZE) {
                    table.delete(deletes);
                    deletes = Lists.newArrayList();
                    counter = 0;
                }
            }
            if (counter > 0)
                table.delete(deletes);
        } catch (Exception e) {
          // ignore
            System.out.println("Error while clean recommendations: " + e.toString());
        } finally {
            TableUtils.closeQuietly(table);
        }                 
        return 0;
    }

}
